package quickstart;

import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * @Author:RenPu
 * @Date: 2020/4/20 14:18
 * @Version: 1.0
 * @description: 编制拓扑任务
 */
public class WordCountApplication {


    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {

        //创建拓扑的构建对象，编制spolt与bolt进行联系起来
        TopologyBuilder builder = new TopologyBuilder();

        //配置LineSpolt的唯一标识为 a1
        builder.setSpout("a1",new LineSpolt());

        //配置LineSplitBolt，的唯一标识为b2，同时设置并行的线程数量以及数据来源那个spolt（通过唯一标识a1）
        builder.setBolt("b2",new LineSplitBolt(),2).shuffleGrouping("a1");

        //配置WordCountBolt，的唯一标识为b3，同时设置并行的线程数量以及数据来源那个bolt（通过唯一标识b2以及元祖信息的唯一标识）
        builder.setBolt("b3",new WordCountBolt(),3).fieldsGrouping("b2",new Fields("word"));




        //创建Config对象
        Config config = new Config();

        //设置当拓扑任务，执行时所需占用的进程数量
        config.setNumWorkers(2);

        //Storm可靠相关性的配置
        config.setNumAckers(0);

        //将Topology任务提交执行。三个参数：1：拓扑任务的名称，配置对象，拓扑对象
        StormSubmitter.submitTopology("storm_wordcount",config,builder.createTopology());



    }

}
